/**
 * This file is part of git-as-svn. It is subject to the license terms
 * in the LICENSE file found in the top-level directory of this distribution
 * and at http://www.gnu.org/licenses/gpl-2.0.html. No part of git-as-svn,
 * including this file, may be copied, modified, propagated, or distributed
 * except according to the terms contained in the LICENSE file.
 */
package svnserver.server;

import org.jetbrains.annotations.NotNull;
import org.mapdb.DB;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.tmatesoft.svn.core.SVNErrorCode;
import org.tmatesoft.svn.core.SVNErrorMessage;
import org.tmatesoft.svn.core.SVNException;
import org.tmatesoft.svn.core.SVNURL;
import svnserver.parser.*;
import svnserver.parser.token.ListBeginToken;
import svnserver.parser.token.ListEndToken;
import svnserver.repository.RepoContainer;
import svnserver.repository.VcsRepository;
import svnserver.repository.VcsRepositoryMapping;
import svnserver.repository.mapping.RepositoryListMapping;
import svnserver.server.command.*;
import svnserver.server.msg.AuthReq;
import svnserver.server.msg.ClientInfo;
import svnserver.server.step.Step;
import svnserver.service.*;
import svnserver.service.Parameters;

import java.io.*;
import java.net.*;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;

/**
 * Сервер для предоставления доступа к git-у через протокол subversion.
 *
 * @author Artem V. Navrotskiy <bozaro@users.noreply.github.com>
 */
public final class SvnServer extends Thread {
    @NotNull
    private static final Logger log = LoggerFactory.getLogger(SvnServer.class);
    private static final long FORCE_SHUTDOWN = TimeUnit.SECONDS.toMillis(5);
    @NotNull
    private final Map<String, BaseCmd<?>> commands = new HashMap<>();
    @NotNull
    static private final Map<Long, Socket> connections = new ConcurrentHashMap<>();
    @NotNull
    static private volatile Map<Long, Long> sessionMap = new ConcurrentHashMap<>();
    @NotNull
    private final VcsRepositoryMapping repositoryMapping;
    @NotNull
    private final DB cacheDb;
    @NotNull
    private final Parameters parameters;
    @NotNull
    private final ServerSocket serverSocket;
    @NotNull
    private final ExecutorService poolExecutor;
    @NotNull
    private final AtomicBoolean stopped = new AtomicBoolean(false);
    @NotNull
    private final AtomicLong lastSessionId = new AtomicLong();

    public SvnServer() throws IOException, SVNException {
        setDaemon(true);
        parameters = Parameters.Get();
        File file = new File(parameters.BasePath().getAbsolutePath() + "/caches");
        if (!file.exists()) {
            if (!file.mkdir()) {
                log.error("create dir: {} failed ", file.getAbsolutePath());
            }
        }
        try {
            cacheDb = Parameters.createCache(file);
        } catch (Exception e) {
            log.error("create cache database", e);
            throw e;
        }
        commands.put("commit", new CommitCmd());
        commands.put("diff", new DeltaCmd(DiffParams.class));
        commands.put("get-locations", new GetLocationsCmd());
        commands.put("get-location-segments", new GetLocationSegmentsCmd());
        commands.put("get-latest-rev", new GetLatestRevCmd());
        commands.put("get-dated-rev", new GetDatedRevCmd());
        commands.put("get-dir", new GetDirCmd());
        commands.put("get-file", new GetFileCmd());
        commands.put("get-iprops", new GetIPropsCmd());
        commands.put("log", new LogCmd());
        commands.put("reparent", new ReparentCmd());
        commands.put("check-path", new CheckPathCmd());
        commands.put("replay", new ReplayCmd());
        commands.put("replay-range", new ReplayRangeCmd());
        commands.put("rev-prop", new RevPropCmd());
        commands.put("rev-proplist", new RevPropListCmd());
        commands.put("stat", new StatCmd());
        commands.put("status", new DeltaCmd(StatusParams.class));
        commands.put("switch", new DeltaCmd(SwitchParams.class));
        commands.put("update", new DeltaCmd(UpdateParams.class));

        commands.put("lock", new LockCmd());
        commands.put("lock-many", new LockManyCmd());
        commands.put("unlock", new UnlockCmd());
        commands.put("unlock-many", new UnlockManyCmd());
        commands.put("get-lock", new GetLockCmd());
        commands.put("get-locks", new GetLocksCmd());

        final RepositoryListMapping.Builder builder = new RepositoryListMapping.Builder();
        repositoryMapping = builder.build();
        serverSocket = new ServerSocket();
        serverSocket.setReuseAddress(true);
        for (int i = 0; i < 10; i++) {
            if (bind(i)) {
                break;
            }
        }

        if (parameters.IsUseCachedThreadPool()) {
            log.info("Use cached ThreadPool");
            poolExecutor = Executors.newCachedThreadPool();
        } else {
            log.info("Use Fixedsize ThreadPool , counts: {}", parameters.Threads());
            poolExecutor = Executors.newFixedThreadPool(parameters.Threads());
        }

        log.info("Server bind: {}", serverSocket.getLocalSocketAddress());
    }

    private boolean bind(int n) {
        try {
            serverSocket.setReuseAddress(true);
            serverSocket
                    .bind(new InetSocketAddress(InetAddress.getByName(parameters.ListenAddress()), parameters.Port()));
        } catch (BindException e) {
            log.error("bind {}:{}", parameters.ListenAddress(), parameters.Port());
            if (n >= 9) {
                log.error("error:", e);
                log.error("sserver is now exiting.");
                System.exit(1);
            }
            return false;
        } catch (Exception e) {
            log.error("error:", e);
            log.error("sserver is now exiting.");
            System.exit(1);
        }
        return true;
    }

    public int getPort() {
        return serverSocket.getLocalPort();
    }

    @Override
    public void run() {
        log.info("Server is ready on port: {}", serverSocket.getLocalPort());
        ScheduledExecutorService scheduledThreadPool = Executors.newScheduledThreadPool(1);
        scheduledThreadPool.scheduleAtFixedRate(new Sweeper(), 0, 120, TimeUnit.SECONDS);
        while (!stopped.get()) {
            final Socket client;
            try {
                client = this.serverSocket.accept();
                client.setSoTimeout(60 * 1000);// ms
            } catch (IOException e) {
                if (stopped.get()) {
                    log.info("Server Stopped");
                    break;
                }
                log.error("Error accepting client connection", e);
                continue;
            }

            long sessionId = lastSessionId.incrementAndGet();
            poolExecutor.execute(() -> {
                log.info("New connection from: {} session: {}", client.getRemoteSocketAddress(), sessionId);
                long starttime = System.currentTimeMillis();
                try (Socket clientSocket = client) {
                    connections.put(sessionId, client);
                    sessionMap.put(sessionId, starttime);
                    serveClient(clientSocket, sessionId);
                } catch (EOFException | SocketException ignore) {
                    // client disconnect is not a error
                } catch (SVNException | IOException e) {
                    log.error("Client error", e);
                } catch (Exception e) {
                    log.error("Exception error", e);
                } finally {
                    connections.remove(sessionId);
                    sessionMap.remove(sessionId);
                    log.info("session: {} closed. total: {} ms", sessionId, (System.currentTimeMillis() - starttime));
                }
            });
        }
    }

    private void serveClient(@NotNull Socket socket, long sessionId) throws IOException, SVNException {
        socket.setTcpNoDelay(true);
        final SvnServerWriter writer = new SvnServerWriter(new BufferedOutputStream(socket.getOutputStream()));
        final SvnServerParser parser = new SvnServerParser(socket.getInputStream());

        final ClientInfo clientInfo = exchangeCapabilities(parser, writer);
        SubversionPath spath = SubversionPath.Builder(clientInfo.getUrl());
        if (spath == null) {
            BaseCmd.sendError(writer,
                    SVNErrorMessage.create(SVNErrorCode.BAD_URL, "Bad URL:" + clientInfo.getUrl().toString()));
            parser.skipItems();
            return;
        }

        if (spath.RefIndex() == ReferenceIndex.ReferenceTags) {
            BaseCmd.sendError(writer, SVNErrorMessage.create(SVNErrorCode.UNSUPPORTED_FEATURE,
                    "The tag operation is not supported, Your download milestone form project page"));
            parser.skipItems();
            return;
        }
        final SubversionContext svnctx;
        if (clientInfo.getUrl().getProtocol().equals("svn+ssh") && Parameters.Get().SshIsAllowed()) {
            svnctx = SubversionContext.Create(parser, writer, clientInfo.getUserAgent(), spath, true);
            if (svnctx == null || svnctx.AuthResult() == null) {
                BaseCmd.sendError(writer,
                        SVNErrorMessage.create(SVNErrorCode.AUTHN_NO_PROVIDER, "no provider auth info"));
                parser.skipItems();
                return;
            }

            Anonymous(parser, writer, clientInfo.getUrl().getHost());

        } else {
            // log.info("use PLAIN auth");
            svnctx = ContextAuthenticator(parser, writer, clientInfo.getUrl().getHost(), spath);
        }

        //

        switch (svnctx.AuthResult().ErrorCode()) {
            case svnserver.service.AuthorizeResult.OK:
                break;
            case svnserver.service.AuthorizeResult.NotFound:
                BaseCmd.sendError(writer,
                        SVNErrorMessage.create(SVNErrorCode.RA_SVN_REPOS_NOT_FOUND, svnctx.AuthResult().Message()));
                parser.skipItems();
                return;
            case svnserver.service.AuthorizeResult.NotAllow:
                BaseCmd.sendError(writer,
                        SVNErrorMessage.create(SVNErrorCode.RA_SVN_REPOS_NOT_FOUND, svnctx.AuthResult().Message()));
                parser.skipItems();
                return;
            case svnserver.service.AuthorizeResult.AuthError:
                BaseCmd.sendError(writer,
                        SVNErrorMessage.create(SVNErrorCode.RA_NOT_AUTHORIZED, svnctx.AuthResult().Message()));
                parser.skipItems();
                return;
            default:
                BaseCmd.sendError(writer, SVNErrorMessage.create(SVNErrorCode.UNKNOWN, svnctx.AuthResult().Message()));
                parser.skipItems();
                return;
        }

        long size = RepositoryUtils.CalculateSize(svnctx.AuthResult().RepositoryPath());
        if (size > svnctx.AuthResult().Rlimit()) {
            String msg = String.format("This Repository size is %s, more than %s. Access deined.",
                    AuthorizeManager.DiskSize(size), AuthorizeManager.DiskSize(svnctx.AuthResult().Rlimit()));
            BaseCmd.sendError(writer, SVNErrorMessage.create(SVNErrorCode.AUTHZ_UNREADABLE, msg));
            parser.skipItems();
            return;
        }
        String branch = spath.Branch();
        if (branch.isEmpty()) {
            branch = RepositoryUtils.DiscoverHead(svnctx.AuthResult().RepositoryPath());
            if (branch == null) {
                BaseCmd.sendError(writer, SVNErrorMessage.create(SVNErrorCode.BAD_URL,
                        "This Repository no default branch. Access deined."));
                parser.skipItems();
                return;
            }
        }

        log.info("session: {} [{}] -> {} reposize: {} limit: {}", sessionId, svnctx.getUser().getEmail(),
                spath.BaseURL(), AuthorizeManager.DiskSize(size),
                AuthorizeManager.DiskSize(svnctx.AuthResult().Rlimit()));

        final SVNURL baseURL = SVNURL.parseURIEncoded(spath.BaseURL());
        RepoContainer repocontainer;
        try {
            repocontainer = repositoryMapping.getRepoContainer(baseURL, cacheDb, svnctx.AuthResult().RepositoryPath(),
                    branch);
        } catch (SVNException e) {
            BaseCmd.sendError(writer, e.getErrorMessage());
            parser.skipItems();
            return;
        }

        if (repocontainer == null) {
            BaseCmd.sendError(writer, SVNErrorMessage.create(SVNErrorCode.RA_SVN_REPOS_NOT_FOUND,
                    "This repository is empty or the branch not exists. url:  " + clientInfo.getUrl()));
            parser.skipItems();
            return;
        }

        final SessionContext context = new SessionContext(parser, writer, this, repocontainer, clientInfo,
                svnctx.getUser());
        final VcsRepository repository = context.getRepository();
        /// bind value
        repository.delayUpdate(spath.PathNode(), svnctx.AuthResult().RIMEID(), svnctx.AuthResult().Flimit());
        try {
            repository.updateRevisions();
        } catch (ClassCastException e) {
            repositoryMapping.EraseRepoContainer(repocontainer);
            log.error("updateRevision ClassCastException \n", e);
            BaseCmd.sendError(writer, SVNErrorMessage.create(SVNErrorCode.RA_SVN_IO_ERROR,
                    "Update revision broken, There may be a file with an illegal file name in this repository."));
            parser.skipItems();
            return;
        } catch (NullPointerException e) {
            repositoryMapping.EraseRepoContainer(repocontainer);
            log.error("NullPointerException\n", e);
            BaseCmd.sendError(writer, SVNErrorMessage.create(SVNErrorCode.RA_SVN_IO_ERROR,
                    "Update revision broken ,We suggest your use git or checkout again !"));
            parser.skipItems();
            return;
        } catch (Exception e) {
            log.error("updateRevision failed \n", e);
            BaseCmd.sendError(writer, SVNErrorMessage.create(SVNErrorCode.RA_SVN_IO_ERROR,
                    "Update revision broken ,We suggest your use git or checkout again !"));
            parser.skipItems();
            return;
        }

        sendAnnounce(writer, repocontainer);

        while (!isInterrupted()) {
            try {
                Step step = context.poll();
                if (step != null) {
                    step.process(context);
                    continue;
                }

                final SvnServerToken token = parser.readToken();
                if (token != ListBeginToken.instance) {
                    throw new IOException("Unexpected token: " + token);
                }
                final String cmd = parser.readText();
                final BaseCmd<?> command = commands.get(cmd);
                if (command != null) {
                    log.info("Repository: {} Receive command: {}", spath.PathNode(), cmd);
                    if (!command.getCommandOnlyRead()) {
                        if (!svnctx.AuthResult().Writable()) {
                            BaseCmd.sendError(writer, SVNErrorMessage.create(SVNErrorCode.AUTHZ_UNWRITABLE,
                                    "Your have not authorized to write this repository"));
                            parser.skipItems();
                            return;
                        }
                    }
                    processCommand(context, command, parser);
                    // if (cmd.equals("update")) {
                    // Resque.SubversionEvent(spath.RepoOwner(), spath.RepoName(), spath.Branch(),
                    // svnctx.getUser().UserId());
                    // }
                } else {
                    log.warn("Unsupported command: {}", cmd);
                    BaseCmd.sendError(writer,
                            SVNErrorMessage.create(SVNErrorCode.RA_SVN_UNKNOWN_CMD, "Unsupported command: " + cmd));
                    parser.skipItems();
                    return;
                }

            } catch (SVNException e) {
                log.error("Command execution error: {}", e.getCause() != null ? e.getCause().getMessage() : e);
                BaseCmd.sendError(writer, e.getErrorMessage());
                return;
            }
        }
    }

    private static <T> void processCommand(@NotNull SessionContext context, @NotNull BaseCmd<T> cmd,
            @NotNull SvnServerParser parser) throws IOException, SVNException {
        final T param = MessageParser.parse(cmd.getArguments(), parser);
        parser.readToken(ListEndToken.class);
        cmd.process(context, param);
    }

    private ClientInfo exchangeCapabilities(SvnServerParser parser, SvnServerWriter writer)
            throws IOException, SVNException {
        // Анонсируем поддерживаемые функции.
        writer.listBegin().word("success").listBegin().number(2).number(2).listBegin().listEnd().listBegin()
                .word("edit-pipeline") // This is required.
                .word("svndiff1") // We support svndiff1
                .word("absent-entries") // We support absent-dir and absent-dir editor commands
                // .word("commit-revprops") // We don't currently have _any_ revprop support
                // .word("mergeinfo") // Nope, not yet
                .word("depth").word("inherited-props") // Need for .gitattributes and .gitignore
                .word("log-revprops") // svn log --with-all-revprops
                .listEnd().listEnd().listEnd();

        // Читаем информацию о клиенте.
        final ClientInfo clientInfo = MessageParser.parse(ClientInfo.class, parser);
        if (clientInfo.getProtocolVersion() != 2) {
            throw new SVNException(SVNErrorMessage.create(SVNErrorCode.VERSION_MISMATCH,
                    "Unsupported protocol version: " + clientInfo.getProtocolVersion() + " (expected: 2)"));
        }
        return clientInfo;
    }

    @NotNull

    private SubversionContext ContextAuthenticator(@NotNull SvnServerParser parser, @NotNull SvnServerWriter writer,
            @NotNull String hostURI, @NotNull SubversionPath spath) throws IOException, SVNException {
        writer.listBegin().word("success").listBegin().listBegin().word(" PLAIN").listEnd()
                .string(parameters.IsFilterDomain() ? parameters.Domain() : hostURI).listEnd().listEnd();
        while (true) {
            // Читаем выбранный вариант авторизации.
            final AuthReq authReq = MessageParser.parse(AuthReq.class, parser);
            if (!authReq.getMech().equals("PLAIN")) {
                sendError(writer, "Unsupported auth type: " + authReq.getMech());
                continue;
            }
            final SubversionContext sctx = SubversionContext.Create(parser, writer, authReq.getToken(), spath, false);
            if (sctx == null || sctx.AuthResult().ErrorCode() == svnserver.service.AuthorizeResult.AuthError) {
                sendError(writer, "incorrect credentials");
                continue;
            }

            writer.listBegin().word("success").listBegin().listEnd().listEnd();

            return sctx;
        }
    }

    @NotNull
    public void Anonymous(@NotNull SvnServerParser parser, @NotNull SvnServerWriter writer, @NotNull String host)
            throws IOException, SVNException {
        // Отправляем запрос на авторизацию.
        writer.listBegin().word("success").listBegin().listBegin().word(" ANONYMOUS").listEnd()
                .string(parameters.IsFilterDomain() ? parameters.Domain() : host).listEnd().listEnd();

        while (true) {
            // Читаем выбранный вариант авторизации.
            final AuthReq authReq = MessageParser.parse(AuthReq.class, parser);
            if (!authReq.getMech().equals("ANONYMOUS")) {
                sendError(writer, "unknown auth type: " + authReq.getMech());
                continue;
            }

            writer.listBegin().word("success").listBegin().listEnd().listEnd();
            return;
        }
    }

    private void sendAnnounce(@NotNull SvnServerWriter writer, @NotNull RepoContainer repoinfo) throws IOException {
        writer.listBegin().word("success").listBegin().string(repoinfo.getRepository().getUuid())
                .string(repoinfo.getBaseUrl().toString()).listBegin()
                // .word("mergeinfo")
                .listEnd().listEnd().listEnd();
    }

    private static void sendError(SvnServerWriter writer, String msg) throws IOException {
        writer.listBegin().word("failure").listBegin().string(msg).listEnd().listEnd();
    }

    public void startShutdown() throws IOException {
        if (stopped.compareAndSet(false, true)) {
            log.info("Shutdown server");
            serverSocket.close();
            poolExecutor.shutdown();
        }
    }

    public void shutdown(long millis) throws InterruptedException, IOException {
        startShutdown();
        if (!poolExecutor.awaitTermination(millis, TimeUnit.MILLISECONDS)) {
            forceShutdown();
        }
        join(millis);
        cacheDb.close();
        log.info("Server shutdown");
    }

    private void forceShutdown() throws IOException, InterruptedException {
        for (Socket socket : connections.values()) {
            socket.close();
        }
        poolExecutor.awaitTermination(FORCE_SHUTDOWN, TimeUnit.MILLISECONDS);
    }

    static class Sweeper implements Runnable {
        @Override
        public void run() {
            log.debug("Scheduled ExecutorService Running");
            log.debug("current session counts: {}.", sessionMap.size());
            sessionMap.forEach((id, beginTime) -> {
                if (System.currentTimeMillis() - beginTime > Parameters.Get().Deadline() * 1000) {
                    log.warn("session: {} timeout. starttime: {}", id, beginTime);
                    try {
                        connections.get(id).close();
                        sessionMap.remove(id);
                    } catch (Exception e) {
                        log.warn("close expires session: {} failed", id, e);
                    }
                }
            });
        }
    }
}
